#!/usr/bin/python
# visualize the task execution time break down
# map: task start, map start, map finish, task finish
# reduce: task start, reduce start, suffle finish, sort finish, reduce finish, task finish

import sys
import re

def getvalue(a):
	return a[a.find("=")+1:].strip("\"")

def savevalue(a, b, c, f):
	len = b - a
	delta = len/2
	center = a + delta
	tmp = str(center)+' '+str(c)+' '+str(delta)+'\n'
	f.write(tmp)



if len(sys.argv) <= 1:
	print "file name?\n"
	sys.exit(-1)

# precess log, get time break down information 
maptasks = {}
reducetasks = {}
job   = {}
start = {}
for l in open(sys.argv[1]):
	A = l.split()
	if A[0] == "Job":
		for a in A:
			# store job information
			if a.startswith("LAUNCH_TIME"):
				job["launch_time"] = int(getvalue(a))
				start = job["launch_time"]
			if a.startswith("FINISH_TIME"):
				job["finish_time"] = int(getvalue(a))
	elif A[0] in ["Task","MapAttempt", "ReduceAttempt"]:
	   	task = {}
	   	status = {}
	   	id = {}
	   	for a in A:
		   	# read task type
		   	if a.startswith("TASK_TYPE"):
			   	task["task_type"] = getvalue(a)

			# read task id
		   	if a.startswith("TASKID"):
			   	id = getvalue(a)
			   	task["taskid"] = int(id[id.rfind("_")+1:])

		   	# read task status
		   	if a.startswith("TASK_STATUS"):
			   	status = getvalue(a)

		   	# read important time spot
		   	if a.startswith("START_TIME"):
			   	if A[0] == "Task":
				   	task["task_start"] = int(getvalue(a))
			   	elif A[0] == "MapAttempt":
				   	task["map_start"] = int(getvalue(a)) 
			   	elif A[0] == "ReduceAttempt":
				   	task["reduce_start"] = int(getvalue(a))

		   	if a.startswith("FINISH_TIME"):
			   	if A[0] == "Task":
				   	task["task_finish"] = int(getvalue(a))
			   	elif A[0] == "MapAttempt":
				   	task["map_finish"] = int(getvalue(a))
			   	elif A[0] == "ReduceAttempt":
				   	task["reduce_finish"] = int(getvalue(a))

		   	if a.startswith("SHUFFLE_FINISHED"):
			   	task["reduce_shuffle"] = int(getvalue(a))

		   	if a.startswith("SORT_FINISHED"):
			   	task["reduce_sort"] = int(getvalue(a)) 

	   	# ignore SETUP and CLEANUP task
	   	if task["task_type"] in ["SETUP", "CLEANUP"]:
		   	continue

	   	# ignore KILLED or FAILED task
	   	if status in ["KILLED","FAILED"]:
		   	continue

	   	if task["task_type"] == "MAP":
		   	if not maptasks.has_key(task["taskid"]):
			   	maptasks[task["taskid"]] = {}
		  	for k,v in task.items():
				# avoid the speculative task attempt overwrite the correct map_start
				if k == "map_start":
					if maptasks[task["taskid"]].has_key("task_finish"):
						continue
			   	maptasks[task["taskid"]][k] = v

	   	if task["task_type"] == "REDUCE":
		   	if not reducetasks.has_key(task["taskid"]):
			   	reducetasks[task["taskid"]] = {}
		   	for k,v in task.items():
				# avoid the speculative task attempt overwrite the correct reduce_start
				if k == "reduce_start":
					if reducetasks[task["taskid"]].has_key("task_finish"):
						continue
			   	reducetasks[task["taskid"]][k] = v

# output time break down information
mf1 = open('map1.data', 'w')
mf2 = open('map2.data', 'w')
mf3 = open('map3.data', 'w')
mf = open('map.data', 'w')
tasklist = maptasks.values()
for task in tasklist:
	savevalue (task["task_start"], task["map_start"], task["taskid"], mf1)
	savevalue (task["map_start"], task["map_finish"], task["taskid"], mf2)
	savevalue (task["map_finish"], task["task_finish"], task["taskid"], mf3)
	mf.write(str(task["taskid"])+' '+ str(task["task_start"]) + ' ' + str(task["map_start"]) + ' ' + str(task["map_finish"]) + ' ' + str(task["task_finish"]) + '\n' )

mf1.close()
mf2.close()
mf3.close()
mf.close()



rf1 = open('reduce1.data', 'w')
rf2 = open('reduce2.data', 'w')
rf3 = open('reduce3.data', 'w')
rf4 = open('reduce4.data', 'w')
rf5 = open('reduce5.data', 'w')
rf = open('reduce.data', 'w')
tasklist = reducetasks.values()
for task in tasklist:
	savevalue (task["task_start"], task["reduce_start"], task["taskid"], rf1)
	savevalue (task["reduce_start"], task["reduce_shuffle"], task["taskid"], rf2)
	savevalue (task["reduce_shuffle"], task["reduce_sort"], task["taskid"], rf3)
	savevalue (task["reduce_sort"], task["reduce_finish"], task["taskid"], rf4)
	savevalue (task["reduce_finish"], task["task_finish"], task["taskid"], rf5)
	rf.write(str(task["taskid"])+' '+ str(task["task_start"]) + ' ' + str(task["reduce_start"]) + ' ' + str(task["reduce_shuffle"]) + ' ' + str(task["reduce_sort"]) + ' ' + str(task["reduce_finish"]) + ' ' + str(task["task_finish"]) + '\n' )


rf1.close()
rf2.close()
rf3.close()
rf4.close()
rf5.close()
rf.close()



#job["life"] = job["finish_time"] - job["launch_time"]
#cmptlist = []
#trackers = {}
#taskslist = tasks.values()
#for task in taskslist:
#	t = task["finish_time"] - task["start_time"]
#   tracker = task["tracker"]
#   task["life"] = t
#   cmptlist.append(t)
#   if  not  trackers.has_key(tracker):
#	   trackers[tracker] = [t]
#   else:
#	   trackers[tracker].append(t)

#print "job completion time:", job["life"]
#print "task completion time: mean %.2f, min %.2f, max %.2f"%(mean(cmptlist), min(cmptlist), max(cmptlist))
#print "per tracker"
#for tracker, tl in trackers.items():
#	print "%s: mean  %.2f,  max %.2f, sum %.2f, len %d"%(repr(tracker), mean(tl), max(tl), sum(tl), len(tl))

